package com.intct.func;

import com.intct.bean.WeightedAvgAccum;
import org.apache.flink.table.functions.AggregateFunction;

import java.util.Iterator;

/**
 * @author gufg
 * @since 2025-10-29 10:56
 */
public class WeightedAvg extends AggregateFunction<Long, WeightedAvgAccum> {

    /*
        调用 AggregateFunction 的 createAccumulator() 方法创建一个空的 accumulator
     */
    @Override
    public WeightedAvgAccum createAccumulator() {
        return new WeightedAvgAccum();
    }

    /*
        数据都处理完了之后，通过调用 getValue 方法来计算和返回最终的结果
     */
    @Override
    public Long getValue(WeightedAvgAccum acc) {
        if (acc.count == 0) {
            return null;
        } else {
            return acc.sum / acc.count;
        }
    }

    /*
        每一行数据，会调用 accumulate() 方法来更新 accumulator
     */
    public void accumulate(WeightedAvgAccum acc, long iValue, int iWeight) {
        acc.sum += iValue * iWeight;
        acc.count += iWeight;
    }

    /*
        在 bounded OVER 窗口中是必须实现的
     */
    public void retract(WeightedAvgAccum acc, long iValue, int iWeight) {
        acc.sum -= iValue * iWeight;
        acc.count -= iWeight;
    }

    /*
        如果聚合函数用在会话窗口（当两个会话窗口合并的时候需要 merge 他们的 accumulator）的话，merge() 方法就是必须要实现的
     */
    public void merge(WeightedAvgAccum acc, Iterable<WeightedAvgAccum> it) {
        Iterator<WeightedAvgAccum> iter = it.iterator();
        while (iter.hasNext()) {
            WeightedAvgAccum a = iter.next();
            acc.count += a.count;
            acc.sum += a.sum;
        }
    }

    /*
        在许多批式聚合中是必须实现
     */
    public void resetAccumulator(WeightedAvgAccum acc) {
        acc.count = 0;
        acc.sum = 0L;
    }
}
